package com.zk.iotgateway.rabbitmq;

import com.zk.iotgateway.domain.param.IotDispatchDO;
import com.rabbitmq.client.BuiltinExchangeType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

/**
 * rabbitmq
 * 生产者
 * 动态绑定
 *
 * @author:mn.m
 * @create 2019-04-28 下午 04:31
 **/
public class PublisherConfig {

    private static final Logger log = LoggerFactory.getLogger(PublisherConfig.class);

    private IotDispatchDO iotDispatchDO;

    private RabbitTemplate rabbitTemplate;


    public PublisherConfig() {
    }

    /*
    * 初始化
    * */
    public void initConfig(IotDispatchDO iotDispatchDO){
        this.iotDispatchDO=iotDispatchDO;
        this.rabbitTemplate = rabbitTemplate();
        RabbitAdmin rabbitAdmin = new RabbitAdmin(this.rabbitTemplate.getConnectionFactory());
        String[] queueNames =this.iotDispatchDO.getQueueName().split(",");
        if (BuiltinExchangeType.DIRECT.getType().equals(this.iotDispatchDO.getExchangeType())){
            DirectExchange directExchange = new DirectExchange(this.iotDispatchDO.getExchange());
            rabbitAdmin.declareExchange(directExchange);
            for (String queueName:queueNames) {
                /*1.队列名称，2.是否可持久化，3.是否排他列队，4.是否自动删除（没有消费者订阅时）*/
                Queue queue=new Queue(queueName);
                rabbitAdmin.declareQueue(queue);
                Binding binding = BindingBuilder.bind(queue).to(directExchange).with(this.iotDispatchDO.getRoutingKey());
                rabbitAdmin.declareBinding(binding);
            }
        }else if (BuiltinExchangeType.TOPIC.getType().equals(this.iotDispatchDO.getExchangeType())){
            TopicExchange topicExchange = new TopicExchange(this.iotDispatchDO.getExchange());
            rabbitAdmin.declareExchange(topicExchange);
            for (String queueName:queueNames) {
                /*1.队列名称，2.是否可持久化，3.是否排他列队，4.是否自动删除（没有消费者订阅时）*/
                Queue queue=new Queue(queueName);
                rabbitAdmin.declareQueue(queue);
                Binding binding = BindingBuilder.bind(queue).to(topicExchange).with(this.iotDispatchDO.getRoutingKey());
                rabbitAdmin.declareBinding(binding);
            }
        }else if (BuiltinExchangeType.FANOUT.getType().equals(this.iotDispatchDO.getExchangeType())){
            FanoutExchange fanoutExchange = new FanoutExchange(this.iotDispatchDO.getExchange());
            rabbitAdmin.declareExchange(fanoutExchange);
            for (String queueName:queueNames) {
                /*1.队列名称，2.是否可持久化，3.是否排他列队，4.是否自动删除（没有消费者订阅时）*/
                Queue queue=new Queue(queueName);
                rabbitAdmin.declareQueue(queue);
                Binding binding = BindingBuilder.bind(queue).to(fanoutExchange);
                rabbitAdmin.declareBinding(binding);
            }
        }/*else {
            HeadersExchange headersExchange = new HeadersExchange(this.iotDispatchDO.getExchange());
            rabbitAdmin.declareExchange(headersExchange);
            //指定headers的匹配类型(all、any)、键值对
            Map<String, Object> headers=new HashMap<String, Object>();
            headers.put("x-match", "any");//all any(只要有一个键值对匹配即可)
            headers.put("key","sssss");
            rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(headersExchange).whereAny(headers).match());

        }*/

    }

    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory(this.iotDispatchDO));
        System.out.println(template.getClass().getName()+"-------------");
       //The routing key is set to the name of the queue by the broker for the default exchange.
       /* template.setRoutingKey(this.iotDispatchDO.getRoutingKey());
        template.setQueue(this.iotDispatchDO.getQueueName());*/
        /*设置转换器*/
       //template.setMessageConverter(new JsonMessageConverter());
        //发布确认
        template.setConfirmCallback( new RabbitTemplate.ConfirmCallback() {
            //消息发送到queue时就执行
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                log.debug(correlationData+"//////");
                if (!b){
                    log.debug("发送到queue失败");
                    throw new RuntimeException("send error " + s);
                }
            }
        });
        return template;

    }

    public ConnectionFactory connectionFactory(IotDispatchDO iotDispatchDO) {
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setAddresses(iotDispatchDO.getAddress());
        cachingConnectionFactory.setUsername(iotDispatchDO.getUsername());
        cachingConnectionFactory.setPassword(iotDispatchDO.getPassword());
        cachingConnectionFactory.setPort(Integer.parseInt(iotDispatchDO.getPort()));
        cachingConnectionFactory.setVirtualHost(iotDispatchDO.getVirtualHost());
        /**
         * 对于每一个RabbitTemplate只支持一个ReturnCallback。
         * 对于返回消息，模板的mandatory属性必须被设定为true，
         * 它同样要求CachingConnectionFactory的publisherReturns属性被设定为true。
         * 如果客户端通过调用setReturnCallback(ReturnCallback callback)注册了RabbitTemplate.ReturnCallback，那么返回将被发送到客户端。
         * 这个回调函数必须实现下列方法：
         *void returnedMessage(Message message, intreplyCode, String replyText,String exchange, String routingKey);
         */
        // connectionFactory.setPublisherReturns(true);
        /**
         * 同样一个RabbitTemplate只支持一个ConfirmCallback。
         * 对于发布确认，template要求CachingConnectionFactory的publisherConfirms属性设置为true。
         * 如果客户端通过setConfirmCallback(ConfirmCallback callback)注册了RabbitTemplate.ConfirmCallback，那么确认消息将被发送到客户端。
         * 这个回调函数必须实现以下方法：
         * void confirm(CorrelationData correlationData, booleanack);
         */
        cachingConnectionFactory.setPublisherConfirms(iotDispatchDO.isPublisherConfirms());
        return cachingConnectionFactory;
    }

    public ConnectionFactory getConn(){
        return this.rabbitTemplate.getConnectionFactory();
    }

    /*
     * 发送消息
     *
     * */
    public void send(String s) {
        this.rabbitTemplate.convertAndSend(s);
    }


    /*
     * 发送消息
     * */
    public void send(String exchange,String routingKey,Object msg) {
        this.rabbitTemplate.convertAndSend(exchange,routingKey,msg);
    }


}
